package com.shujia.mr.wc1;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapTask;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.task.JobContextImpl;
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;

import java.io.IOException;

/*
    编写mapreduce程序的时候，需要单独编写一个map类和reduce类以及一个运行入口类
 */
// 编写Map类
/*
    自定义一个类，继承Hadoop中的Mapper类，重写map方法

hello world java
hadoop nihao
shijie hello world jaba
java
hive hadoop hbase spark flink
hello world
shijie hello world jaba
 */
//Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
//输入：<0L, "hello world">
//输出：<"hello",1L>  <"world",1L>
class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

    //每一个map任务执行之前执行setup方法
    @Override
    protected void setup(Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        //创建数据库连接资源
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        // 这里的重写方法逻辑，是每一行数据都要执行一遍的
        String line = value.toString(); // "hello world"
        String[] words = line.split(" ");

        //context: hadoop运行时的上下文对象
        for (String word : words) {
            context.write(new Text(word), new LongWritable(1L));
        }
    }

    @Override
    protected void cleanup(Mapper<LongWritable, Text, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        //释放setup中创建的数据库连接资源
    }
}
//环形缓冲区，溢写，排序都不需要自己编写
//开发时只需要关心，map以及reduce的计算逻辑即可

/*
    编写Reduce类
 */
//Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
    @Override
    protected void reduce(Text key, Iterable<LongWritable> values, Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException {
        long sum = 0L;
        for (LongWritable value : values) {
            long l = value.get();
            sum+=l;
        }
        context.write(key, new LongWritable(sum));
    }
}


public class MRDemo1 {
    public static void main(String[] args) throws Exception{
        //获取Hadoop环境配置对象
        Configuration conf = new Configuration();

        //设置主节点
        conf.set("fs.defaultFS", "hdfs://master:9000");
//        conf.set("mapreduce.job.inputformat.class","KeyValueTextInputFormat");
//        conf.set("mapreduce.input.fileinputformat.split.minsize","256L");
//        conf.set("mapreduce.input.fileinputformat.split.maxsize","64L");
        //环形缓冲区溢写比例
//        conf.set("mapreduce.map.sort.spill.percent","0.8");
        //环形缓冲区大小设置
//        conf.set("mapreduce.task.io.sort.mb","200");

        //创建Job作业
        Job job = Job.getInstance(conf);

        job.setJarByClass(MRDemo1.class);

        //设置当前job作业的名字
        job.setJobName("32期 单词统计案例mapreduce实现");

        //设置当前作业要执行的map类
        job.setMapperClass(MyMapper.class);

        //设置Combiner要执行的类，一般情况下和reduce是一个类
        job.setCombinerClass(MyReducer.class);


        //设置当前作业要执行的reduce类
        job.setReducerClass(MyReducer.class);

        //设置Map任务输出的键值类型和上面指定的map类输出键值类型一一对应
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);

        //设置Reduce任务输出的键值类型和上面指定的reduce类输出键值类型一一对应
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        // 设置数据读取的路径【hdfs上的路径】
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        // 设置数据输出路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));


        // 提交作业到yarn上
        boolean b = job.waitForCompletion(true);
        if(b){
            System.out.println("32期 单词统计案例mapreduce实现执行成功！>_-");
        }else {
            System.out.println("32期 单词统计案例mapreduce实现执行失败！T_T");
        }

//        MapTask
//            jobContext = new JobContextImpl(job, id, reporter);
//            taskContext = new TaskAttemptContextImpl(job, taskId, reporter);
//        outputFormat:
//        TextInputFormat
//        MapTask.MapOutputBuffer
    }
}
/*
    1、将程序打包，放到linux环境中执行
    2、执行命令 hadoop jar xxxx.jar 主类名路径 输入路径 输出路径
    3、在yarn的页面中查看提交状态 master:8088
    4、在hdfs上查看结果
 */

/**
 *  计算split切片公式：
 *      blockSize：128
 *      minSize：1
 *      maxSize：2^63-1
 *
 *      Math.max(minSize, Math.min(maxSize, blockSize)) = 64
 *
 *
 *  分区编号产生的公式：
 *      (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks
 */
